package com.shujia.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

object Demo16Mayi {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("mayi")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._


    //用户流水数据
    val ant_user_low_carbon: DataFrame = spark.read
      .format("csv")
      .option("sep", "\t")
      .schema("user_id STRING,data_dt STRING,low_carbon DOUBLE")
      .load("data/ant_user_low_carbon.txt")


    //植物表
    val ant_plant_carbon: DataFrame = spark.read
      .format("csv")
      .option("sep", "\t")
      .schema("plant_id STRING,plant_name STRING,plant_carbon DOUBLE")
      .load("data/ant_plant_carbon.txt")


    /**
     * 蚂蚁森林植物申领统计
     * 问题：假设2017年1月1日开始记录低碳数据（user_low_carbon），假设2017年10月1日之前满足申领条件的用户都申领了一颗p004-胡杨，
     * 剩余的能量全部用来领取“p002-沙柳” 。
     * 统计在10月1日累计申领“p002-沙柳” 排名前10的用户信息；以及他比后一名多领了几颗沙柳。
     * 得到的统计结果如下表样式：
     *
     */

    //取出胡杨和沙柳所需要的能量
    val huyangplantCarbon: Double = ant_plant_carbon
      .where($"plant_name" === "胡杨")
      .select($"plant_carbon")
      .rdd
      .collect()
      .map { case Row(low_carbon: Double) => low_carbon }
      .head

    val shaliuplantCarbon: Double = ant_plant_carbon
      .where($"plant_name" === "沙柳")
      .select($"plant_carbon")
      .rdd
      .collect()
      .map { case Row(low_carbon: Double) => low_carbon }
      .head


    ant_user_low_carbon
      //取出2017年1月1号到201710月1号的数据
      .where($"data_dt" >= "2017/1/1" and $"data_dt" <= "2017/10/1")
      //计算每个用户总的减少碳排放量
      .groupBy($"user_id")
      .agg(sum($"low_carbon") as "sum_low_carbon")
      //满申领条件的用户都领取了一颗胡杨
      .withColumn("shengyu_low_carbon", when($"sum_low_carbon" >= huyangplantCarbon, $"sum_low_carbon" - huyangplantCarbon).otherwise($"sum_low_carbon"))
      //计算剩余的能量还能领取多少沙柳
      .withColumn("num", floor($"shengyu_low_carbon" / shaliuplantCarbon))
      //获取后一名领取的棵树
      .withColumn("last_num", lead($"num", 1, 0) over Window.orderBy($"num".desc))
      //计算多领取的棵树
      .withColumn("duo", $"num" - $"last_num")
      //取前十
      .limit(10)
      .show()


    /**
     *
     * 蚂蚁森林低碳用户排名分析
     * 问题：查询user_low_carbon表中每日流水记录，条件为：
     * 用户在2017年，连续三天（或以上）的天数里，
     * 每天减少碳排放（low_carbon）都超过100g的用户低碳流水。
     * 需要查询返回满足以上条件的user_low_carbon表中的记录流水。
     * 例如用户u_002符合条件的记录如下，因为2017/1/2~2017/1/5连续四天的碳排放量之和都大于等于100g：
     *
     */

    ant_user_low_carbon
      //取2017年的数据
      .where(substring($"data_dt", 0, 4) === "2017")
      //计算每天减少的碳排放量
      .groupBy($"user_id", $"data_dt")
      .agg(sum($"low_carbon") as "sum_low_carbon")
      //过滤掉不到100的数据
      .where($"sum_low_carbon" > 100)
      //计算连续
      .withColumn("r", row_number() over Window.partitionBy($"user_id").orderBy($"data_dt"))
      //使用日期减去排名，判断是否连续
      .withColumn("clazz", expr("date_sub(regexp_replace(data_dt, '/', '-'),r)"))
      //计算连续的天数
      .withColumn("lianxu", count($"user_id") over Window.partitionBy($"user_id", $"clazz"))
      //取出连续3天以上的数据
      .where($"lianxu" >= 3)
      //关联原表获取流水
      .join(ant_user_low_carbon, List("user_id", "data_dt"))
      .select($"user_id", $"data_dt", $"low_carbon")
      .show(100)


  }

}
